package com.zengjianjun.canal.demo.consumer;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.zengjianjun.canal.demo.annotation.CanalListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author zengjianjun
 */
@Slf4j
@Component
public class CanalConsumer implements CanalConsumerInterface {

    @CanalListener(destination = "user", subscribe = "py_web.user", threadNum = 4, batchSize = 2000)
    public void userConsumer(Message message) {
        try {
            this.messageProcessor(message);
        } catch (Exception e) {
            log.info("canal message process error: ", e);
        }
    }

//    @CanalListener(destination = "test", subscribe = "py_web.test")
    public void testConsumer(Message message) {
        try {
            this.messageProcessor(message);
        } catch (Exception e) {
            log.info("canal message process error: ", e);
        }
    }

    private void messageProcessor(Message message) throws InvalidProtocolBufferException {
        // 数据过滤
        List<CanalEntry.Entry> list = this.filterInsertEntry(message);
        // 数据解析
        int size = list.size();
        int batchSize = 0;
        for (int i = 0; i < size; i++) {
            CanalEntry.Entry entry = list.get(i);
            // 打印当前批次最后一条Header信息
            if (i == size - 1) {
                log.info(String.format("binlog[%s:%s] , dbName:%s, tableName:%s ",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName()));
            }
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
            batchSize += rowDataList.size();
            for (int j = 0; j < rowDataList.size(); j++) {
                CanalEntry.RowData rowData = rowDataList.get(j);
                List<CanalEntry.Column> columnList = rowData.getAfterColumnsList();
                Map<String, CanalEntry.Column> columnMap = columnList.stream().collect(Collectors.toMap(CanalEntry.Column::getName, Function.identity()));
                CanalEntry.Column idColum = columnMap.get("id");
                CanalEntry.Column usernameColumn = columnMap.get("username");
                CanalEntry.Column passwordColumn = columnMap.get("password");
                CanalEntry.Column createTimeColumn = columnMap.get("create_time");
                // 打印当前批次总共处理的数据行数及最后一行数据的信息
                if (j == rowDataList.size() - 1 && i == size - 1) {
                    log.info("batchSize = {}, id = {}, createTime = {}", batchSize, this.getColumnValue(idColum), this.getColumnValue(createTimeColumn));
                }
            }
        }
    }
}
